package com.hkbigdata.ontimer;

import com.hkbigdata.entry.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author liuanbo
 * @creat 2023-05-17-17:21
 * @see 2194550857@qq.com
 * ws_1001,1684121131000,3
 * ws_1001,1684121139000,2
 * ws_1001,1684121140000,1
 * ws_1001,1684121142000,4
 * ws_1001,1684121147000,5
 * ws_1001,1684121148000,6
 */
public class Flink01_ProcessingTimce_OnTimer {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取端口数据并转换为JavaBean
        SingleOutputStreamOperator<WaterSensor> waterSensorDS = env.socketTextStream("hadoop102", 9999)
                .map(data -> {
                    String[] split = data.split(",");
                    return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                });


        SingleOutputStreamOperator<WaterSensor> process = waterSensorDS.keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, WaterSensor>() {
                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {

                        //1.获取当前时间
                        long ts = ctx.timerService().currentProcessingTime();
                        System.out.println("当前时间:" + ts);

                        //2.注册五秒钟的时间定时器
                        long TimerTs = ts + 5000L;
                        System.out.println("定时器注册时间:" + TimerTs);
                        ctx.timerService().registerProcessingTimeTimer(TimerTs);

                        //3.输出数据
                        out.collect(value);

                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<WaterSensor> out) throws Exception {
                        System.out.println("定时器触发时间" + timestamp);
                    }
                });

        process.print();

        env.execute();
    }
}
